package com.duowan.realtime.tool;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.IOUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import com.duowan.common.util.DateConvertUtils;
import com.duowan.datawarehouse.util.FileUtil;
import com.duowan.datawarehouse.utils.DateUtil;
import com.duowan.realtime.computing.BloomFilterClient;
import com.duowan.realtime.thirft.api.BloomFilterException;
import com.duowan.realtime.thirft.api.BloomFilterGroupQuery;
import com.duowan.realtime.util.MapUtils;


public class LoadDataToBloomfilterPoolTool {
	
	public String bloomfilterGroup;

	private static final Logger LOG = LoggerFactory.getLogger(LoadDataToBloomfilterPoolTool.class);
	
	private BloomFilterClient client;
	
	long result = 0L;
	
	private static final String DEFAULT_PARTITION = "19700101";
	
	public LoadDataToBloomfilterPoolTool(){
		client = new BloomFilterClient();
	}
	
	public static void main(String[] args) throws IOException {
		if (args.length != 2) {
			LOG.error("must be 2 parameter: bloomfilterGroup,filePath");
			System.exit(-1);
		}
		
		LoadDataToBloomfilterPoolTool tool = new LoadDataToBloomfilterPoolTool();
		tool.bloomfilterGroup = args[0];
		String filePath = args[1];
		String partition = args.length<=2?DEFAULT_PARTITION:args[2];
		tool.excuteLoad(filePath,partition);
		
	}

	private void excuteLoad(String filePath,String partition) {
		List<File> fileList = FileUtil.listFiles(filePath);
		long startTime = System.currentTimeMillis();
		long lineCounter = 0L;
		LOG.info("there are " + fileList.size() + " files prepare to load to bloomfilter pool");
		
		for (File file : fileList) {
			LOG.info("load file :" + file.getName());
			Map<String,List<String>> queryMap = new HashMap<String, List<String>>();
			BufferedReader bufferedReader = null;
			try {
				bufferedReader = new BufferedReader(new FileReader(file));
				String line = bufferedReader.readLine();
				while (isNotBlank(line)) {
					allotQueryByGroup(queryMap, line);
					line = bufferedReader.readLine();
					
					lineCounter++;
				}
				excuteThriftBloomfilter(queryMap,partition);
				
			} catch (Exception e) {
				LOG.error("excute history data to bloomfilter error", e);
			} finally{
				IOUtils.closeQuietly(bufferedReader);
			}
		}
		long endTime = System.currentTimeMillis();
		LOG.info("records:" + lineCounter + ",cost time:" + (endTime-startTime)/1000 + ",load history data tps:" + (lineCounter/(endTime - startTime) * 1000));
	}
	
	public void excuteThriftBloomfilter(Map<String,List<String>> queryMap,String partition) throws BloomFilterException, TException{
		List<BloomFilterGroupQuery> bfQueryList = new LinkedList<BloomFilterGroupQuery>();
		Set<String> keySet = queryMap.keySet();
		for (String key : keySet) {
			bfQueryList.add(initBfQuery(queryMap, key, partition));
		}
		Map<String, Integer> result = client.notContainsCountAndAdd(bloomfilterGroup, bfQueryList);
	}

	private BloomFilterGroupQuery initBfQuery(Map<String, List<String>> queryMap, String key, String partition) {
		BloomFilterGroupQuery query = new BloomFilterGroupQuery();
		query.setGroup(key);
		query.setPartition(partition);
		query.setValues(queryMap.get(key));
		return query;
	}

	protected String parseKey(String key) {
		String[] keyArray = key.split("/");
		String dateKey = DateConvertUtils.extract(DateConvertUtils.parse(keyArray[1], "yyyy-MM-dd"), "yyyyMMdd").toString();
		if ((keyArray.length & 1) != 0) {
			return dateKey;
		}
		return dateKey + "/" + keyArray[1];
	}

	protected void allotQueryByGroup(Map<String, List<String>> queryMap,String line) {
		Map map = convertLineToMap(line);
		if (map.size() == 0 ) {
			return;
		}
		//fieldSplit[0]=group fieldSplit[1]=value
		if (queryMap.containsKey(map.get("key"))) {
			if (isNotBlank((String)map.get("value"))) {
				queryMap.get(map.get("key")).add((String)map.get("value"));
			}
		}else {
			List<String> values = new LinkedList<String>();
			String value = (String)map.get("value");
			if (isNotBlank(value)) {
				values.add(value);
			}
			queryMap.put((String)map.get("key"), values);
		}
	}

	private boolean isNotBlank(String value) {
		return org.apache.commons.lang.StringUtils.isNotBlank(value);
	}
	
	protected Map convertLineToMap(String line) {
		String[] columns = splitLine(line);
		Map map = MapUtils.convertToMap(columns, "key","value");
		return map;
	}
	
	protected String[] splitLine(String line) {
		return StringUtils.trimArrayElements(line.split(","));
	}
	
}
